package com.imooc.dianping.canal;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.protobuf.InvalidProtocolBufferException;
import com.imooc.dianping.dal.ShopModelMapper;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * canal 消息消费模型
 *
 * @author gblfy
 * @date 2021-11-22
 */
@Component
public class CanalScheduling implements Runnable, ApplicationContextAware {

    //记录日志
    private final static Logger logger = LoggerFactory.getLogger(CanalScheduling.class);
    private ApplicationContext applicationContext;

    @Autowired
    private ShopModelMapper shopModelMapper;
    @Resource
    private CanalConnector canalConnector;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    @Override
    //每个100毫秒 唤醒线程执行run方法
    @Scheduled(fixedDelay = 100)
    public void run() {
        //初始化批次ID
        long batchId = -1;
        try {
            //批次/1000条
            int batchSize = 1000;
            Message message = canalConnector.getWithoutAck(batchSize);
            //1000条批次的ID  当获取的batchId=-1代表没有消息
            batchId = message.getId();
            List<CanalEntry.Entry> entries = message.getEntries();

            // batchId != -1 (内部消费有消息的)
            // entries.size() > 0有对应的内容
            //当batchId != -1 并且entries.size() > 0说明mysql bin_log发生了多少条数据的变化
            if (batchId != -1 && entries.size() > 0) {
                //逐条处理
                entries.forEach(entry -> {
                    //处理类型: bin_log已ROW方式处理的消息
                    if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                        //消息解析处理
                        publishCanalEvent(entry);
                    }
                });
            }
            //消息解析完毕后，告知批次消息已经消费ack确认完成
            canalConnector.ack(batchId);
        } catch (Exception e) {
            e.printStackTrace();
            //将本次消息回滚，下次继续消息
            canalConnector.rollback(batchId);
        }
    }

    /**
     * 消息解析处理函数
     *
     * @param entry
     */
    private void publishCanalEvent(CanalEntry.Entry entry) {
        //事件类型 只关注INSERT、UPDATE、DELETE
        CanalEntry.EventType eventType = entry.getHeader().getEventType();
        //获取发生变化的数据库
        String database = entry.getHeader().getSchemaName();
        //获取发生变化的数据库中的表
        String table = entry.getHeader().getTableName();
        CanalEntry.RowChange change = null;
        try {
            //记录这条消息发生了那些变化
            change = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
            return;
        }
        change.getRowDatasList().forEach(rowData -> {
            List<CanalEntry.Column> columns = rowData.getAfterColumnsList();
            //主键
            String primaryKey = "id";
            CanalEntry.Column idColumn = columns.stream().filter(column -> column.getIsKey()
                    && primaryKey.equals(column.getName())).findFirst().orElse(null);

            //将Columns转换成map
            Map<String, Object> dataMap = parseColumnsToMap(columns);
            try {
                indexES(dataMap, database, table);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });
    }

    /**
     * 将Columns转换成map
     *
     * @param columns
     * @return
     */
    Map<String, Object> parseColumnsToMap(List<CanalEntry.Column> columns) {
        Map<String, Object> jsonMap = new HashMap<>();
        columns.forEach(column -> {
            if (column == null) {
                return;
            }
            jsonMap.put(column.getName(), column.getValue());
        });
        return jsonMap;
    }


    private void indexES(Map<String, Object> dataMap, String database, String table) throws IOException {
        logger.info("发生变化的行数据:{} ", dataMap);
        logger.info("发生变化的数据库:{}  ", database);
        logger.info("发生变化的表:{} ", table);

        // 限定处理出具库范围 支处理数据库名称为dianpingdb的消息
        if (!StringUtils.equals("dianpingdb", database)) {
            return;
        }
        /**
         * 我们要关注表数据范围
         * 当seller表、category表、shop 表发生变化，都buildESQuery
         * 当着3个参数中任意一个参数发生变化，我只需要将发生变化的ID传入，重建与此ID关联的索引
         */
        List<Map<String, Object>> result = new ArrayList<>();
        if (StringUtils.equals("seller", table)) {
            result = shopModelMapper.buildESQuery(new Integer((String) dataMap.get("id")), null, null);
        } else if (StringUtils.equals("category", table)) {
            result = shopModelMapper.buildESQuery(null, new Integer((String) dataMap.get("id")), null);
        } else if (StringUtils.equals("shop", table)) {
            result = shopModelMapper.buildESQuery(null, null, new Integer((String) dataMap.get("id")));
        } else {
            //不关注其他的表
            return;
        }

        for (Map<String, Object> map : result) {
            IndexRequest indexRequest = new IndexRequest("shop");
            indexRequest.id(String.valueOf(map.get("id")));
            indexRequest.source(map);
            //更新索引
            restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
        }


    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}
